package com.learn.kafka.transaction;


import com.learn.kafka.config.ConsumerProperties;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.time.Duration;
import java.util.Arrays;

/**
 * 事务消息消费者（继承通用配置父类）
 * @author: learn.kafka
 */
public class TransactionConsumer extends ConsumerProperties {


    public static void main(String[] args) {
        new TransactionConsumer().startTransactionMessages();
    }
    // 构造方法：初始化事务消费特殊配置
    public TransactionConsumer() {
        super(); // 调用父类构造方法，加载通用配置
        initTransactionConsumerConfig();
    }

    // 初始化事务消费者特殊配置
    private void initTransactionConsumerConfig() {
        // 设置消费组ID（事务消息专用）
        setProperty(ConsumerConfig.GROUP_ID_CONFIG, "mate-test-group1");
        // 隔离级别：只消费已提交的事务消息（关键配置）
        setProperty(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");
        // 如需修改集群地址，可在此处覆盖
        // setConfig("bootstrap.servers", "192.168.44.160:9092");
    }

    // 启动消费逻辑
    private void startTransactionMessages() {
        // 从父类获取消费者实例
        KafkaConsumer<Integer, String> consumer = createConsumer();

        // 订阅事务消息主题
        consumer.subscribe(Arrays.asList("transaction-test"));

        try {
            // 持续消费消息
            while (true) {
                ConsumerRecords<Integer, String> records = consumer.poll(Duration.ofMillis(1000));
                for (ConsumerRecord<Integer, String> record : records) {
                    System.out.printf(
                            "offset = %d, key = %s, value = %s, partition = %d%n",
                            record.offset(), record.key(), record.value(), record.partition()
                    );
                }
            }
        } finally {
            // 关闭消费者
            consumer.close();
        }
    }

}
